package com.intct.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.intct.hbase.bean.LocusBean;
import com.intct.common.Constant;
import com.intct.util.KafkaUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author gufg
 * @since 2025-07-28 11:26
 * 处理车辆轨迹行为数据
 */
public class DwdLocusApp {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置并行度
        env.setParallelism(1);

        // 启动检查点
        env.enableCheckpointing(6000L, CheckpointingMode.EXACTLY_ONCE);

        // 配置源 -- ods kafka
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(Constant.KAFKA_ODS_LOCUS_TOPIC, "dwd-locus-source");
        DataStreamSource<String> sourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "dwd-locus-source");

        // 数据处理
        /*
            11,17592719043682,1475977126,104.04276,30.69286
            end,17592719043682
         */
        SingleOutputStreamOperator<String> mapDS = sourceDS.map(new MapFunction<String, String>() {
            @Override
            public String map(String jsonStr) throws Exception {
                // {"op":"r","name":"locus","after":{"driver_id":"11","lng":"104.05187","time":"1475976829","order_id":"17592719043682","lat":"30.70226"},"ts_ms":1753406924046}
                // 转换为JSONObject
                JSONObject jsonObj = JSONObject.parseObject(jsonStr);

                // 获取数据--司机ID(end代表订单，非end代表正常的订单轨迹)
                String driverId = jsonObj.getJSONObject(Constant.MYSQL_CDC_JSON_KEY_AFTER).getString("driver_id");

                JSONObject afterObj = jsonObj.getJSONObject(Constant.MYSQL_CDC_JSON_KEY_AFTER);

                // TODO 规范
                // TODO 1 重复数据

                // TODO 2 空值

                // TODO 3 脱敏

                // TODO 4 单位

                // 格式转换, 生成JavaBean
                LocusBean.LocusBeanBuilder locusBeanBuilder = LocusBean.builder();
                LocusBean locusBean = locusBeanBuilder
                        .driverId(driverId)
                        .orderId(afterObj.getString("order_id"))
                        .build();
                // 如某个订单结束
                if (!"end".equals(driverId)) {
                    locusBean = locusBeanBuilder
                            .driverId(driverId)
                            .orderId(afterObj.getString("order_id"))
                            .time(afterObj.getLong("time"))
                            .lng(afterObj.getString("lng"))
                            .lat(afterObj.getString("lat"))
                            .build();
                }

                // 将JavaBean转为JSON字符串
                return JSON.toJSONString(locusBean);
            }
        });

        // TODO 退维(宽表)

        // 输出 Kafka dwd
        KafkaSink<String> kafkaSink = KafkaUtil.getKafkaSink(Constant.KAFKA_DWD_LOCUS_TOPIC, "dwd_locus_kafka");
        mapDS.sinkTo(kafkaSink);

        // 启动作业
        env.execute();
    }
}
